package org.srilankaliving.analytics;


import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.mapred.lib.NLineInputFormat;


import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.util.Tool;

public class CorelationAnalyzer extends Configured implements Tool  {


    public static class QoLMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
        private String town, category, fileTag = "CD~";

        public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
            //taking one line/record at a time and parsing them into key value pairs
            String line = value.toString();
            String splitarray[] = line.split("\t");
            town = splitarray[0].trim();
            category = splitarray[1].trim();

            //sending the key value pair out of mapper
            output.collect(new Text(town), new Text(fileTag + category));
        }
    }


    public static class EthnicityTownMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {

        private String town, qolCode, fileTag = "DR~";

        public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
            //taking one line/record at a time and parsing them into key value pairs
            String line = value.toString();
            String splitarray[] = line.split("\t");
            town = splitarray[0].trim();
            qolCode = splitarray[1].trim();

            //sending the key value pair out of mapper
            output.collect(new Text(town), new Text(fileTag + qolCode));
        }
    }


    public static class EthnicQoLReducer extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
        private String qolType, ethinicity;

        public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
            while (values.hasNext()) {
                String currValue = values.next().toString();
                String valueSplitted[] = currValue.split("~");
             /*identifying the record source that corresponds to a cell number
             and parses the values accordingly*/
                if (valueSplitted[0].equals("CD")) {
                    qolType = valueSplitted[1].trim();
                } else if (valueSplitted[0].equals("DR")) {
                    //getting the delivery code and using the same to obtain the Message
                    ethinicity = valueSplitted[1].trim();
                }
            }

            if (qolType != null && ethinicity != null) {
                output.collect(new Text(ethinicity), new Text(qolType));
            } /*else if (qolType == null) {
                output.collect(new Text("qolType"), new Text(ethinicity));
            } else if (ethinicity == null) {
                output.collect(new Text(qolType), new Text("ethinicity"));
            }*/
        }

    }

    public int run(String[] args) throws Exception {

        /*Copy Required files from previos MPRed jobs */
        QOLUtils.copyFile(QOLUtils.ETHINIC_GR_OUT_PATH + File.separator + "part-00000",
                QOLUtils.COORELATION_ETH + File.separator + "part-00000");
        QOLUtils.copyFile(QOLUtils.WALLTYPE_OUT_PATH + File.separator + "part-00000",
                QOLUtils.COORELATION_QOL + File.separator + "part-00000");


        JobConf conf = new JobConf(CorelationAnalyzer.class);
        conf.setJobName("CorelationAnalyzer");

        //setting key value types for mapper and reducer outputs
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(Text.class);

        //specifying the custom reducer class
        conf.setReducerClass(EthnicQoLReducer.class);

        //Specifying the input directories(@ runtime) and Mappers independently for inputs from multiple sources
        MultipleInputs.addInputPath(conf,
                new Path(QOLUtils.COORELATION_QOL),
                TextInputFormat.class, QoLMapper.class);
        MultipleInputs.addInputPath(conf,
                new Path(QOLUtils.COORELATION_ETH),
                TextInputFormat.class, EthnicityTownMapper.class);

        //Specifying the output directory @ runtime
        FileOutputFormat.setOutputPath(conf, new Path(QOLUtils.COORELATION_OUT_PATH));

        JobClient.runJob(conf);
        return 0;
    }




}
